package com.atguigu.ct.common.bean;

import com.atguigu.ct.common.api.Column;
import com.atguigu.ct.common.api.RowKey;
import com.atguigu.ct.common.api.TableRef;
import com.atguigu.ct.common.constant.Names;
import com.atguigu.ct.common.constant.ValueConstant;
import com.atguigu.ct.common.util.DateUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.NamespaceNotFoundException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;

/**
 * Description: 基础数据访问对象
 * @Author: yyx
 * @Create: 2023/4/14 16:27
 * @Version: 8
 */
public abstract class BaseDao {
    private ThreadLocal<Connection> connHolder = new ThreadLocal<Connection>();//放在线程中，可以不用每次都获取
    private ThreadLocal<Admin> adminHolder = new ThreadLocal<Admin>();

    protected void start() throws Exception{
        getConnection();
        getAdmin();
    }

    protected void end() throws Exception{
        Admin admin = getAdmin();
        if(admin != null){
            admin.close();
            adminHolder.remove();
        }

        Connection connection = getConnection();
        if(connection != null){
            connection.close();
            connHolder.remove();
        }
    }

    /**
     * 创建表，如果表已经存在，那么删除后再创建新的
     * @param name
     * @param families
     */
    protected void createTableXX(String name,String... families) throws Exception {
        createTableXX(name,null,null,families);
    }
    protected void createTableXX(String name,String coprocessorClass,Integer regionCount,String... families) throws Exception {//方法重载
        Admin admin = getAdmin();
        TableName tableName = TableName.valueOf(name);
        if (admin.tableExists(tableName)){
            //表存在，删除表
            deleteTable(name);
        }
        //创建表
        createTable(name,coprocessorClass,regionCount,families);
    }

    // 1.创建表
    private void createTable(String name,String coprocessorClass,Integer regionCount,
                             String... families) throws Exception {
        Admin admin = getAdmin();
        TableName tableName = TableName.valueOf(name);
        //TableDescriptor对象通过TableDescriptorBuilder构建；
        TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tableName);//表格描述器
        if(families == null || families.length == 0){
            families = new String[1];
            families[0] = Names.CF_INFO.getValue();
        }
        for (String family : families) {
            ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();//构建列族对象
            tableDescriptor.setColumnFamily(columnFamilyDescriptor);//设置列族
        }
        //协处理器判断
        if(coprocessorClass != null && !"".equals(coprocessorClass)){
            tableDescriptor.setCoprocessor(coprocessorClass);
        }
        //增加预分区
        if(regionCount == null || regionCount <= 1){
            admin.createTable(tableDescriptor.build());
        }else{
            byte[][] splitKeys = genSplitKeys(regionCount);//分区键
            admin.createTable(tableDescriptor.build(),splitKeys);
        }
    }
    // 2.生成分区键：避免数据热点，数据倾斜
    private byte[][] genSplitKeys(int regionCount){
        int splitKeyCount = regionCount - 1;
        byte[][] bs = new byte[splitKeyCount][];
        // 0|,1|,2|,3|,4| (|表示ASCII码中第二大的值)
        // (-无穷,0|),(0|,1|),(1|,+无穷)
        List<byte[]> bsList = new ArrayList<byte[]>();
        for(int i = 0;i < splitKeyCount;i++){
            String splitKey = i + "|";
            bsList.add(Bytes.toBytes(splitKey));
        }
        //Collections.sort(bsList,new Bytes.ByteArrayComparator());//排序
        bsList.toArray(bs);//集合数组放到二维数组中
        return bs;
    }
    // 3.删除表
    protected void deleteTable(String name) throws Exception {
        Admin admin = getAdmin();
        TableName tableName = TableName.valueOf(name);
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
    }
    // 4.生成分区号
    protected int genRegionNum(String tel,String data){
        //18103446608拿到最后四位
        String userCode = tel.substring(tel.length()-4);
        //202305010000拿到年和月
        String yearMonth = data.substring(0,6);
        int userCodeHash = userCode.hashCode();
        int yearMonthHash = yearMonth.hashCode();
        // crc校验采用异或算法(不相等是1，相等是0)
        int crc = Math.abs(userCodeHash ^ yearMonthHash);//取绝对值
        // 取模
        int regionNum = crc % ValueConstant.REGION_COUNT;
        return regionNum;
    }

    /**
     * 获取查询时startrow，stoprow集合
     * @return
     */
    protected List<String[]> getStartStorRowkeys(String tel,String start,String end){
        List<String[]> rowkeyss = new ArrayList<String[]>();
        String startTime = start.substring(0,6);
        String endTime = end.substring(0,6);
        // 转换为日期对象，方便计算。例如：2019年8月到2020年3月
        Calendar startCal = Calendar.getInstance();
        startCal.setTime(DateUtil.parse(startTime,"yyyyMM"));
        Calendar endCal = Calendar.getInstance();
        endCal.setTime(DateUtil.parse(endTime,"yyyyMM"));
        while (startCal.getTimeInMillis() <= endCal.getTimeInMillis()){
            //当前月份
            String nowTime = DateUtil.format(startCal.getTime(),"yyyyMM");
            int regionNum = genRegionNum(tel,nowTime);
            // 1_181_202304 ~ 1_181_202304|
            String startRow = regionNum + "_" + tel + "_" + nowTime;
            String stopRow = startRow + "|";
            String[] rowkeys = {startRow,stopRow};
            rowkeyss.add(rowkeys);
            //月份+1
            startCal.add(Calendar.MONTH,1);
        }
        return rowkeyss;
    }

    /**
     * 创建命名空间，如果已经存在，不需要创建，否则创建新的
     * @param namespace
     */
    protected void createNamespaceNX(String namespace) throws Exception {
        Admin admin = getAdmin();
        try {
            admin.getNamespaceDescriptor(namespace);
        }catch (NamespaceNotFoundException e){
            //e.printStackTrace();
            NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(namespace).build();
            admin.createNamespace(namespaceDescriptor);
        }
    }

    /**
     * 增加数据
     * @param name
     * @param put
     */
    protected void putData(String name,Put put) throws Exception {
        // 获取表对象
        Connection connection = getConnection();
        Table table = connection.getTable(TableName.valueOf(name));
        // 增加数据
        table.put(put);
        // 关闭表
        table.close();
    }
    protected void putData(String name,List<Put> puts) throws Exception {
        // 获取表对象
        Connection connection = getConnection();
        Table table = connection.getTable(TableName.valueOf(name));
        // 增加数据
        table.put(puts);
        // 关闭表
        table.close();
    }

    /**
     * 增加对象：自动封装数据，将对象数据直接保存到hbase中去
     * @param obj
     * @throws Exception
     */
    protected void putData(Object obj) throws Exception {
        // 反射
        Class clazz = obj.getClass();
        TableRef tableRef = (TableRef)clazz.getAnnotation(TableRef.class);
        String tableName = tableRef.value();
        Field[] fs = clazz.getDeclaredFields();
        String stringRowkey = "";
        for (Field f : fs) {
            RowKey rowKey = f.getAnnotation(RowKey.class);
            if(rowKey != null){
                f.setAccessible(true);
                stringRowkey = (String)f.get(obj);
                break;
            }
        }
        // 获取表对象
        Connection connection = getConnection();
        Table table = connection.getTable(TableName.valueOf(tableName));
        // 增加数据
        Put put = new Put(Bytes.toBytes(stringRowkey));
        for (Field f : fs) {
            Column column = f.getAnnotation(Column.class);
            if(column != null){
                String family = column.family();
                String colName = column.column();
                if(colName == null || "".equals(colName)){
                    colName = f.getName();
                }
                f.setAccessible(true);
                String value = (String)f.get(obj);
                put.addColumn(Bytes.toBytes(family),Bytes.toBytes(colName),Bytes.toBytes(value));
            }
        }
        table.put(put);
        // 关闭表
        table.close();
    }

    /**
     * 获取连接对象
     * @return
     */
    protected synchronized Connection getConnection() throws Exception{//protected使得只能是它的子类使用
        Connection connection = connHolder.get();
        if(connection == null){
            Configuration conf = HBaseConfiguration.create();
            connection = ConnectionFactory.createConnection(conf);
            connHolder.set(connection);
        }
        return connection;
    }

    /**
     * 获取管理对象
     * @return
     */
    protected synchronized Admin getAdmin() throws Exception{//使得只能是它的子类使用
        Admin admin = adminHolder.get();
        if(admin == null){
            admin = getConnection().getAdmin();
            adminHolder.set(admin);
        }
        return admin;
    }
}
